package com.foogui.demo.produce;

import com.foogui.demo.vo.OrderStep;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;
import java.util.concurrent.TimeUnit;

public class SequenceProducer {

    public static void main(String[] args) throws Exception {
        //1.创建消息生产者producer，并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("produce_group");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("localhost:9876");
        //3.启动producer
        producer.start();
        //订单消息的集合
        List<OrderStep> orderSteps = OrderStep.buildOrders();
        for (int i = 0; i < orderSteps.size(); i++) {
            //4.创建消息对象，指定主题Topic、Tag和消息体
            String body = orderSteps.get(i) + "";
            Message msg = new Message("OrderTopic", "Order", "i" + i , body.getBytes());
            //5.发送同步消息,默认会负载均衡，发送到topic分区的不同队列
//            SendResult result = producer.send(msg);
            /**
             * 参数1：消息
             * 参数2:消息队列的选择器
             * 参数3:选择队列的业务表示（订单id)
             */
            SendResult result =producer.send(msg, new MessageQueueSelector() {
                /**
                 * @param mqs 队列集合
                 * @param msg 消息
                 * @param arg 业务标识
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long orderId = (Long) arg;
                    //将订单号对队列长度取模，就可保证同一订单号一定在同一个队列里
                    long index = orderId % mqs.size();
                    return mqs.get((int) index);
                }
            },  orderSteps.get(i).getOrderId());
            //发送状态
            SendStatus status = result.getSendStatus();
            System.out.println("发送结果：" + result);

            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        producer.shutdown();
    }
}
